package newStreaming

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object updateState extends App {
  val sparkConf = new SparkConf()
  sparkConf.setAppName("wordCountDS")
  sparkConf.setMaster("local[4]")
  val sparkContext = new SparkContext(sparkConf)
  val ssc = new StreamingContext(sparkContext, Durations.seconds(5))
  ssc.checkpoint("data/updateState")
  // 先去node1 启动 nc -lk 9999
  val ds = ssc.socketTextStream("node2", 9999)

  val func = (seq: Seq[Int], opt: Option[Int]) => {
    //计算batch当前结果
    val sum = seq.sum
    //加上之前batch的结果
    val newSum = opt.getOrElse(0) + sum
    Option(newSum)
  }
  // 慎用
  ds.flatMap(_.split(",")).map((_, 1)).updateStateByKey(func).print()


  ssc.start()
  ssc.awaitTermination()
  ssc.stop()
}
